package cn.kennylee.codehub.mq.lip.integration.base;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONUtil;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.lang.NonNull;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 抽象MQ组件，Producer工具
 * <p>Created on 2024/12/16.</p>
 *
 * @param <T> 消息发送结果回调
 * @author kennylee
 * @since 0.0.1
 */
@Getter
@Slf4j
@RequiredArgsConstructor
public abstract class AbstractMqChannel<T> implements MqChannel<T> {

    public static final String DEFAULT_LOCAL_TEST_SCOPE_PROFILE = "local";

    /**
     * 主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     */
    private final String channel;

    /**
     * 同步发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @return 消息发送结果
     */
    @Override
    public T sendMsg(@NonNull String routeKey, @NonNull MessageVo messageVo) {
        log.info("发送消息：channel={}, routeKey={}, messageVo={}", getChannel(), routeKey, convertToMsgJsonStr(messageVo));

        if (isSkipSend()) {
            log.info("测试环境，不发送消息：routeKey={}, messageVo={}", routeKey, convertToMsgJsonStr(messageVo));
            return null;
        }

        return sendMsg(getChannel(), routeKey, messageVo);
    }

    /**
     * 同步发送延迟消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return 消息发送结果
     */
    @Override
    public T sendMsg(@NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        return sendMsg(getChannel(), routeKey, messageVo, delayTime);
    }

    /**
     * 事务提交后，发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> sendMsgByAfterTx(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo) {
        log.info("事务提交后，发送消息：channel={}, routeKey={}, messageVo={}", channel, routeKey, JSONUtil.toJsonStr(messageVo));
        return asyncDoAfterCommit(() -> sendMsg(channel, routeKey, messageVo));
    }

    /**
     * 事务提交后，发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> sendMsgByAfterTx(@NonNull String routeKey, @NonNull MessageVo messageVo) {
        return sendMsgByAfterTx(getChannel(), routeKey, messageVo);
    }

    /**
     * 事务提交后，发送延迟消息
     *
     * @param channel   主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> sendMsgByAfterTx(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        log.info("事务提交后，发送消息：channel={}, routeKey={}, messageVo={}, 延迟时间: {}s", channel, routeKey, JSONUtil.toJsonStr(messageVo), delayTime.getSeconds());
        return asyncDoAfterCommit(() -> sendMsg(channel, routeKey, messageVo, delayTime));
    }

    /**
     * 事务提交后，发送延迟消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> sendMsgByAfterTx(@NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        return sendMsgByAfterTx(getChannel(), routeKey, messageVo, delayTime);
    }

    /**
     * 立即异步发送消息
     *
     * @param channel   主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo) {
        log.info("异步发送消息：channel={}, routeKey={}, messageVo={}", channel, routeKey, JSONUtil.toJsonStr(messageVo));
        return CompletableFuture.supplyAsync(() -> {
            try {
                return ThreadUtil.execAsync(() -> sendMsg(channel, routeKey, messageVo)).get();
            } catch (InterruptedException | ExecutionException e) {
                log.error(e.getMessage(), e);
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 立即异步发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsg(@NonNull String routeKey, @NonNull MessageVo messageVo) {
        return asyncSendMsg(getChannel(), routeKey, messageVo);
    }

    /**
     * 立即异步延迟发送消息
     *
     * @param channel   主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        log.info("异步发送消息：channel={}, routeKey={}, messageVo={}, 延迟时间: {}s", channel, routeKey, JSONUtil.toJsonStr(messageVo), delayTime.getSeconds());
        return CompletableFuture.supplyAsync(() -> {
            try {
                return ThreadUtil.execAsync(() -> sendMsg(channel, routeKey, messageVo, delayTime)).get();
            } catch (InterruptedException | ExecutionException e) {
                log.error(e.getMessage(), e);
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 立即异步延迟发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsg(@NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        return asyncSendMsg(getChannel(), routeKey, messageVo, delayTime);
    }

    /**
     * 事务提交后，异步发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     */
    @Override
    public CompletableFuture<T> asyncSendMsgAfterTx(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo) {
        if (log.isInfoEnabled()) {
            log.info("事务提交后，异步发送消息：channel={}, routeKey={}, messageVo={}", channel, routeKey, JSONUtil.toJsonStr(messageVo));
        }

        return asyncDoAfterCommit(() -> sendMsg(channel, routeKey, messageVo));
    }

    /**
     * 事务提交后，异步发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     */
    @Override
    public CompletableFuture<T> asyncSendMsgAfterTx(@NonNull String routeKey, @NonNull MessageVo messageVo) {
        return asyncSendMsgAfterTx(getChannel(), routeKey, messageVo);
    }

    /**
     * 事务提交后，异步延迟发送消息
     *
     * @param channel   主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsgAfterTx(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        if (log.isInfoEnabled()) {
            log.info("事务提交后，异步发送消息：channel={}, routeKey={}, messageVo={}, 延迟时间: {}s", channel, routeKey, JSONUtil.toJsonStr(messageVo), delayTime.getSeconds());
        }

        return asyncDoAfterCommit(() -> sendMsg(channel, routeKey, messageVo, delayTime));
    }

    /**
     * 事务提交后，异步延迟发送消息
     *
     * @param routeKey  二级路由标识（RocketMQ-Tag / RabbitMQ-RoutingKey / Kafka-SubTopic）
     * @param messageVo 消息内容
     * @param delayTime 延迟时间
     * @return CompletableFuture
     */
    @Override
    public CompletableFuture<T> asyncSendMsgAfterTx(@NonNull String routeKey, @NonNull MessageVo messageVo, Duration delayTime) {
        return asyncSendMsgAfterTx(getChannel(), routeKey, messageVo, delayTime);
    }

    /**
     * 转换为消息JSON字符串
     *
     * @param messageVo 消息内容
     * @return 消息JSON字符串
     */
    protected static String convertToMsgJsonStr(@NonNull MessageVo messageVo) {
        if (Objects.isNull(messageVo.getSendMsgDatetime())) {
            messageVo.setSendMsgDatetime(SystemClock.now());
        }
        return JSONUtil.toJsonStr(messageVo, getJsonConfig());
    }

    /**
     * 构建消息
     *
     * @param messageVo 消息内容
     * @return 消息
     */
    protected static Message<String> buildMessage(@NonNull MessageVo messageVo) {
        Map<String, Object> headers = new HashMap<>(6);
        if (Objects.nonNull(messageVo.getHeaders())) {
            messageVo.getHeaders().forEach((key, value) -> {
                if (StrUtil.isAllNotBlank(key, value)) {
                    headers.put(key, value);
                }
            });
        }

        // 通过消息头(Headers)来传递 Trace ID
        if (!headers.containsKey(MqChannel.TRACE_ID_KEY)) {
            String traceId = MDC.get(MqChannel.TRACE_ID_KEY);
            if (StrUtil.isNotBlank(traceId)) {
                headers.put(MqChannel.TRACE_ID_KEY, traceId);
            }
        }

        MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(convertToMsgJsonStr(messageVo));
        if (!headers.isEmpty()) {
            messageBuilder.copyHeaders(headers);
        }
        return messageBuilder.build();
    }


    /**
     * 获取本地测试环境的profile name
     *
     * @return 本地测试环境的profile name
     */
    protected String getLocalTestScopeProfile() {
        return DEFAULT_LOCAL_TEST_SCOPE_PROFILE;
    }

    private String generateDefaultTopic() {
        return SpringUtil.getApplicationName();
    }

    /**
     * 是否本地测试环境
     *
     * @return 如果是，返回true，否则，返回false
     */
    protected boolean isLocalTestScope() {
        return Set.of(SpringUtil.getActiveProfiles()).contains(getLocalTestScopeProfile());
    }

    /**
     * 是否压力测试
     *
     * @return 如果是，返回true，否则，返回false
     */
    protected boolean isStressTesting() {
        return false;
    }

    /**
     * 是否跳过发送
     *
     * @return 如果是，返回true，否则，返回false
     */
    protected boolean isSkipSend() {
        return isLocalTestScope() || isStressTesting();
    }

    /**
     * 事务提交后异步执行操作（若不在事务中则立即执行）
     *
     * @param commit 需要执行的操作
     * @return CompletableFuture 用于获取异步结果
     * @throws NullPointerException 如果 commit 为 null
     */
    static <T> CompletableFuture<T> asyncDoAfterCommit(final Commit<T> commit) {
        Objects.requireNonNull(commit, "提交动作不能为null");

        CompletableFuture<T> future = new CompletableFuture<>();

        String traceId = MDC.get(TRACE_ID_KEY);

        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            // 事务中的异步执行
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    log.debug("事务已提交，开始异步执行操作");
                    ThreadUtil.execAsync(() -> {
                        if (StrUtil.isNotBlank(traceId)) {
                            // 在事务提交后设置Trace ID
                            MDC.put(TRACE_ID_KEY, traceId);
                        }
                        try {
                            future.complete(commit.doCommit());
                        } catch (Exception e) {
                            log.error("事务提交后异步执行失败", e);
                            future.completeExceptionally(e);
                        } finally {
                            // 确保在异步操作完成后清理Trace ID
                            future.whenComplete((result, throwable) -> {
                                // 清理Trace ID
                                MDC.remove(TRACE_ID_KEY);
                            });
                        }
                    });
                }

                @Override
                public void afterCompletion(int status) {
                    if (status == STATUS_ROLLED_BACK) {
                        log.warn("事务已回滚，取消异步操作");
                        future.cancel(true);
                    }
                }
            });
        } else {
            // 非事务环境立即执行
            log.warn("当前无活跃事务，立即同步执行操作");
            try {
                future.complete(commit.doCommit());
            } catch (Exception e) {
                log.error("同步执行操作失败", e);
                future.completeExceptionally(e);
                // // 非事务环境直接抛出异常
                throw e;
            }
        }

        return future;
    }

    private static JSONConfig getJsonConfig() {
        return JSONConfig.create()
            .setIgnoreNullValue(true)
            .setIgnoreError(true)
            .setDateFormat(DatePattern.NORM_DATETIME_PATTERN);
    }
}
